use bytes::BytesMut;
use log::{error, info};
use std::{collections::HashMap, fs::{self, read_dir}, io::SeekFrom, path::PathBuf};
use tokio::{
    fs::{create_dir_all, File, OpenOptions},
    io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, BufWriter},
    sync::broadcast::Receiver,
    time::Instant,
};

use crate::{Error, SendMessage, DEFAULT_FILENAME, MQ_PATH, SEGMENT_SIZE};

pub struct WalReader {
    pub segment_map: HashMap<String, SegmentReader>,
}

impl WalReader {
    pub fn new() -> Self {
        WalReader {
            segment_map: HashMap::new(),
        }
    }

    pub async fn read_message(
        &mut self,
        topic: &str,
        partitions: &Vec<u16>,
        len: usize,
    ) -> Result<Vec<Message>, Error> {
        let mut messages = Vec::new();
        let start_time = Instant::now();
        //println!("{} {:?} {}", topic, partitions, len);
        loop {
            for partition in partitions {
                if len == messages.len() || start_time.elapsed().as_millis() >= 100 {
                    return Ok(messages);
                }
                let key = format!("{}_{}", topic, partition);
                //info!("{} {} {}, {:?}", topic, partition, key, &self.segment_map);

                let segment_reader = match self.segment_map.get_mut(&key) {
                    Some(v) => v,
                    None => {
                        let file_name = match WalReader::find_last_offset_file(topic, *partition) {
                            Ok(file_name) => file_name,
                            Err(_) => DEFAULT_FILENAME.to_owned(),
                        };
                        let segment_reader = SegmentReader::new(
                            format!("{}/{}/{}", MQ_PATH, topic, partition).into(),
                            file_name,
                        )
                        .await;
                        self.segment_map.insert(key.clone(), segment_reader);
                        let value = self.segment_map.get_mut(&key).unwrap();
                        value
                    }
                };

                // let segment_reader = self.segment_map.entry(key).or_insert(
                //     SegmentReader::new(
                //         format!("{}/{}/{}", MQ_PATH, topic, partition).into(),
                //         "0000000000000000".to_string(),
                //     )
                //     .await
                // );

                match segment_reader.read_message().await {
                    Ok(message) => {
                        messages.push(message);
                    }
                    Err(_e) => {
                        let index = match segment_reader.curr_file_name.parse::<u32>() {
                            Ok(num) => num + 1,
                            Err(_e) => 0,
                        };
                        let file_name = format!("{:016}", index);
                        if fs::metadata(format!(
                            "{}/{}/{}/{}",
                            MQ_PATH, topic, partition, file_name
                        ))
                        .is_ok()
                        {
                            segment_reader.curr_file_name = file_name;
                            let segment = Segment::new(
                                format!("{}/{}/{}", MQ_PATH, topic, partition).into(),
                                segment_reader.curr_file_name.as_str(),
                            )
                            .await;
                            let (offset_file, position) = SegmentReader::read_offset_file(format!("{}/{}/{}", MQ_PATH, topic, partition).into(), &segment_reader.curr_file_name).await?;
                            segment_reader.offset_file = BufWriter::new(offset_file);
                            segment_reader.buf_reader = BufReader::new(segment.open_file().await?);
                            segment_reader.segment = segment;
                            segment_reader.position = position;
                            if let Ok(message) = segment_reader.read_message().await {
                                messages.push(message);
                            }
                        }
                    }
                };
            }
        }
        //Ok(messages)
    }

    fn find_last_offset_file(topic: &str, partition: u16) -> Result<String, Error>{
        let path: PathBuf = format!("{}/{}/{}", MQ_PATH, topic, partition).into();
        if !path.is_dir() {
            return Err(Error::CustomError("The path is not a directory.".to_owned()));
        }

        let mut largest = 0;
        for entry in read_dir(path)? {
            let path = entry?.path();
            if !path.is_file() {
                continue;
            }
            if let Some(filename) = path.file_name() {
                let filename = filename.to_string_lossy();
                let index = match filename.rfind("."){
                    Some(index) => index,
                    None => continue,
                };
                if ".offset" != &filename[index..] {
                    continue;
                }

                let filename = &filename[..index];
                let name = match filename.parse::<u32>() {
                    Ok(name) => name,
                    Err(_) => 0,
                };

                largest = std::cmp::max(largest, name);
                
            }
        }

        Ok(format!("{:016}", largest))
    }
}

pub struct WalWriter {
    pub segment_map: HashMap<String, SegmentWriter>,
}

impl WalWriter {
    pub fn new(receiver: Receiver<SendMessage>) {
        let mut wal = WalWriter {
            segment_map: HashMap::new(),
        };
        tokio::spawn(async move {
            if let Err(e) = wal.persisted(receiver).await {
                error!("{}", e);
                panic!("wal persisted error");
            }
        });
        //wal
    }

    pub async fn write_message(&mut self, message: SendMessage) -> Result<(), Error> {
        let key = format!("{}_{}", &message.topic, message.partition);
        // let segment_writer = self.segment_map.entry(key).or_insert(
        //     SegmentWriter::new(
        //         format!("{}/{}/{}", MQ_PATH, &message.topic, message.partition).into(),
        //         0,
        //     )
        //     .await,
        // );
        let segment_writer = match self.segment_map.get_mut(&key) {
            Some(value) => value,
            None => {
                let segment_writer = SegmentWriter::new(
                    format!("{}/{}/{}", MQ_PATH, &message.topic, message.partition).into(),
                    0,
                )
                .await;
                self.segment_map.insert(key.clone(), segment_writer);
                let value = self.segment_map.get_mut(&key).unwrap();
                value
            }
        };

        if let Ok(chunk_offset) = segment_writer
            .write_message(SegmentMessage::new(message.buf, message.timestamp))
            .await
        {
            if chunk_offset >= SEGMENT_SIZE {
                let index = match segment_writer.curr_file_name.parse::<u32>() {
                    Ok(num) => num + 1,
                    Err(_e) => 0,
                };
                segment_writer.curr_file_name = format!("{:016}", index);
                let segment = Segment::new(
                    format!("{}/{}/{}", MQ_PATH, &message.topic, message.partition).into(),
                    segment_writer.curr_file_name.as_str(),
                )
                .await;
                segment_writer.block_number = segment_writer.block_number + 1;
                segment_writer.chunk_offset = 0;
                segment_writer.buf_writer = BufWriter::new(segment.open_file().await?);
                segment_writer.segment = segment;
            }
        }
        Ok(())
    }

    pub async fn persisted(&mut self, mut receiver: Receiver<SendMessage>) -> Result<(), Error> {
        loop {
            if let Ok(msg) = receiver.recv().await {
                //info!("receive message: {:?}", msg);
                self.write_message(msg).await?;
            }
        }
        //Ok(())
    }
}

#[derive(Debug)]
pub struct Segment {
    pub path: PathBuf,
}

impl Segment {
    pub async fn new(mut path: PathBuf, file_name: &str) -> Self {
        path.push(file_name);

        if let Some(dir) = path.parent() {
            if let Err(_) = create_dir_all(dir).await {
                panic!("create dir error");
            };
        }

        Segment { path }
    }

    pub async fn open_file(&self) -> Result<File, Error> {
        let file = match OpenOptions::new()
            .create(true) // 如果文件不存在则创建
            .write(true) // 可写模式
            .read(true)
            .open(self.path.as_path())
            .await
        {
            Ok(file) => file,
            Err(_) => return Err(Error::CustomError("create segment file error".to_owned())),
        };
        Ok(file)
    }
}

pub struct SegmentWriter {
    pub segment: Segment,
    pub curr_file_name: String,
    pub buf_writer: BufWriter<File>,
    pub block_number: u32,
    pub chunk_offset: u64,
}

impl SegmentWriter {
    pub async fn new(path: PathBuf, block_number: u32) -> Self {
        let file_name = largest_file(&path).unwrap();
        let segment = Segment::new(path, file_name.as_str()).await;
        let file = match segment.open_file().await {
            Ok(file) => file,
            Err(_) => panic!("create segment file error"),
        };
        SegmentWriter {
            segment,
            curr_file_name: file_name,
            buf_writer: BufWriter::new(file),
            block_number,
            chunk_offset: 0,
        }
    }

    pub async fn write_message(&mut self, mut message: SegmentMessage) -> Result<u64, Error> {
        self.buf_writer.seek(SeekFrom::End(0)).await?;

        let len = message.buf.len() as u32;
        self.buf_writer.write_u32(len).await?;
        self.buf_writer.write_u128(message.timestamp).await?;
        self.buf_writer.write_buf(&mut message.buf).await?;
        self.buf_writer.flush().await?;

        self.chunk_offset = self.chunk_offset + len as u64 + 16 + 4;

        Ok(self.chunk_offset)
    }
}

#[derive(Debug)]
pub struct SegmentReader {
    pub segment: Segment,
    pub curr_file_name: String,
    pub buf_reader: BufReader<File>,
    pub position: u64,
    pub offset_file: BufWriter<File>,
}

impl SegmentReader {
    pub async fn new(path: PathBuf, file_name: String) -> Self {
        let segment = Segment::new(path.clone(), file_name.as_str()).await;
        let file = match segment.open_file().await {
            Ok(file) => file,
            Err(_) => panic!("create segment file error"),
        };

        let (offset_file, position) = SegmentReader::read_offset_file(path, &file_name)
            .await
            .expect("create offset file error");
        //println!("{:?}, {}", offset_file, position);

        SegmentReader {
            segment,
            curr_file_name: file_name,
            buf_reader: BufReader::new(file),
            position,
            offset_file: BufWriter::new(offset_file),
        }
    }

    pub async fn read_offset_file(
        mut path: PathBuf,
        file_name: &str,
    ) -> Result<(File, u64), Error> {
        path.push(format!("{}.offset", file_name));
        let (offset_file, position) = match OpenOptions::new()
            .create(true) // 如果文件不存在则创建
            .write(true) // 可写模式
            .read(true)
            .open(path)
            .await
        {
            Ok(mut file) => {
                let position = match file.read_u64().await {
                    Ok(p) => p,
                    Err(_) => 0,
                };
                (file, position)
            }
            Err(_) => panic!("create offset file error"),
        };
        Ok((offset_file, position))
    }

    pub async fn read_message(&mut self) -> Result<Message, Error> {
        self.buf_reader.seek(SeekFrom::Start(self.position)).await?;
        let len: u32 = self.buf_reader.read_u32().await?;
        let timestamp: u128 = self.buf_reader.read_u128().await?;
        let mut buf = BytesMut::with_capacity(len as usize);
        self.buf_reader.read_buf(&mut buf).await?;
        self.position = self.position + len as u64 + 16 + 4;

        self.record_offset(self.position).await?;

        Ok(Message::new(len, timestamp, buf))
    }

    pub async fn record_offset(&mut self, position: u64) -> Result<(), Error> {
        self.offset_file.seek(SeekFrom::Start(0)).await?;
        self.offset_file.write_u64(position).await?;
        self.offset_file.flush().await?;
        Ok(())
    }
}

#[derive(Debug)]
pub struct Message {
    pub len: u32,
    pub timestamp: u128,
    pub msg: BytesMut,
}

impl Message {
    pub fn new(len: u32, timestamp: u128, msg: BytesMut) -> Self {
        Message {
            len,
            timestamp,
            msg,
        }
    }
}

pub struct SegmentMessage {
    pub timestamp: u128,
    pub buf: BytesMut,
}

impl SegmentMessage {
    pub fn new(buf: BytesMut, timestamp: u128) -> Self {
        SegmentMessage { timestamp, buf }
    }
}

fn largest_file(dir_path: &PathBuf) -> Result<String, Error> {
    if !dir_path.exists() {
        // 创建路径及其所有中间目录
        fs::create_dir_all(dir_path)?;
        info!("Path created: {:?}", dir_path);
    }

    let entries = fs::read_dir(dir_path)?;
    // 遍历目录条目
    let mut largest_file_name = DEFAULT_FILENAME.to_string();
    let mut largest = 0;
    for entry in entries {
        let entry = entry?;
        let path = entry.path();

        if path.is_file() {
            if let Some(file_name) = path.file_name() {
                let file_name = file_name.to_string_lossy().to_string();
                let index = match file_name.parse::<u32>() {
                    Ok(num) => num + 1,
                    Err(_e) => 0,
                };
                if index > largest {
                    largest = index;
                    largest_file_name = file_name;
                }
            }
        }
    }
    Ok(largest_file_name)
}
